package com.isoftstone.storage.fastdfs;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.Serializable;

import org.codehaus.plexus.util.IOUtil;
import org.csource.common.NameValuePair;
import org.csource.fastdfs.DownloadCallback;
import org.csource.fastdfs.StorageClient;
import org.csource.fastdfs.StorageClient1;
import org.csource.fastdfs.StorageServer;
import org.csource.fastdfs.TrackerClient;
import org.csource.fastdfs.TrackerServer;
import org.csource.fastdfs.UploadCallback;

import com.isoftstone.storage.AbstractStorage;
import com.isoftstone.storage.ConnectionException;
import com.isoftstone.storage.LazyFileOutputStream;
import com.isoftstone.storage.ResourceDoesNotExistException;
import com.isoftstone.storage.Storage;
import com.isoftstone.storage.TransferFailedException;
import com.isoftstone.storage.authentication.AuthenticationException;
import com.isoftstone.storage.authorization.AuthorizationException;
import com.isoftstone.storage.events.TransferEvent;
import com.isoftstone.storage.resource.Resource;

public class FastDfsStorage extends AbstractStorage implements Storage {
	public static final String NAME_SEPARATOR = ".";
	private TrackerServer trackerServer;

	@Override
	public void get(String resourceName, File destination)
			throws TransferFailedException, ResourceDoesNotExistException,
			AuthorizationException {
		Resource resource = new Resource(resourceName);
		fireGetInitiated(resource, destination);
		OutputStream output = null;
		try {
			fireTransferDebug("attempting to create parent directories for destination: "
					+ destination.getName());
			output = new LazyFileOutputStream(destination);
			downdloadByStream(resource, output);
			finishGetTransfer(resource, output);
		} catch (IOException e) {
			if (destination.exists()) {
				boolean deleted = destination.delete();
				if (!deleted) {
					destination.deleteOnExit();
				}
			}
			fireTransferError(resource, e, TransferEvent.REQUEST_GET);
			String msg = "GET request of: " + resource.getName() + " from "
					+ repository + " failed";
			throw new TransferFailedException(msg, e);
		} finally {
			IOUtil.close(output);
			cleanupGetTransfer(resource);
		}
	}

	@Override
	public FastDfsResourceName put(File source, String resourceName)
			throws TransferFailedException, ResourceDoesNotExistException,
			AuthorizationException {
		resourceName = "to FastDFS-" + source.getName();
		Resource resource = new Resource(resourceName);
		firePutInitiated(resource, source);
		resource.setContentLength(source.length());
		InputStream input = null;
		try {
			firePutStarted(resource, source);
			long length = source.length();
			input = new FileInputStream(source);
			String[] result = uploadFileByStream(input, resource,
					source.getCanonicalPath(), length);
			firePutCompleted(resource, source);
			return new FastDfsResourceName(result);
		} catch (FileNotFoundException e) {
			fireTransferError(resource, e, TransferEvent.REQUEST_PUT);
			throw new TransferFailedException(
					"Specified source file does not exist: " + source, e);
		} catch (IOException e) {
			fireTransferError(resource, e, TransferEvent.REQUEST_PUT);
			String msg = "PUT request to: " + resource.getName() + " in "
					+ repository + " failed";
			throw new TransferFailedException(msg, e);
		} finally {
			IOUtil.close(input);
		}

	}

	protected void finishGetTransfer(Resource resource, OutputStream output)
			throws TransferFailedException {
	}

	@Override
	public void delete(String resourceName) throws TransferFailedException,
			AuthorizationException {
		try {
			FastDfsResourceName name = FastDfsResourceName.parse(resourceName);
			getClient().delete_file(name.getGroupName(),
					name.getRemoteFilename());
		} catch (Exception e) {
			throw new TransferFailedException(e.getMessage());
		}
	}

	@Override
	protected void openConnectionInternal() throws ConnectionException,
			AuthenticationException {
		try {
			TrackerClient tracker = new TrackerClient();
			trackerServer = tracker.getConnection();
		} catch (IOException e) {
			throw new ConnectionException(e.getMessage());
		}
	}

	@Override
	protected void closeConnection() throws ConnectionException {
		if (trackerServer != null) {
			try {
				trackerServer.close();
			} catch (IOException e) {
				throw new ConnectionException(e.getMessage());
			}
		}
	}

	public void downdloadByStream(Resource resource, OutputStream output)
			throws IOException {
		TransferEvent transferEvent = new TransferEvent(this, resource,
				TransferEvent.TRANSFER_PROGRESS, TransferEvent.REQUEST_GET);
		try {
			FastDfsResourceName name = FastDfsResourceName.parse(resource
					.getName());
			getClient().download_file(name.getGroupName(),
					name.getRemoteFilename(),
					new DownloadFileReceiver(resource, output, transferEvent));
		} catch (Exception e) {
			throw new IOException("get file \"" + resource.getName()
					+ "\"fails");
		}
	}

	/**
	 * Upload File to DFS, directly transferring java.io.InputStream to
	 * java.io.OutStream
	 * 
	 * @param fileBuff
	 *            file to be uploaded.
	 * @param uploadFileName
	 *            the name of the file.
	 * @param fileLength
	 *            the length of the file.
	 * @return the file ID in DFS.// results[0]: groupName, results[1]:
	 *         remoteFilename.
	 * @throws IOException
	 */
	public String[] uploadFileByStream(InputStream input, Resource resource,
			String uploadFileName, long fileLength) throws IOException {
		String[] results = null;
		String fileExtName = "";
		if (uploadFileName.contains(".")) {
			fileExtName = uploadFileName.substring(uploadFileName
					.lastIndexOf(".") + 1);
		} else {
			throw new IOException(
					"Fail to upload file, because the format of filename is illegal.");
		}
		NameValuePair[] metaList = new NameValuePair[3];
		metaList[0] = new NameValuePair("fileName", uploadFileName);
		metaList[1] = new NameValuePair("fileExtName", fileExtName);
		metaList[2] = new NameValuePair("fileLength",
				String.valueOf(fileLength));

		try {
			TransferEvent transferEvent = new TransferEvent(this, resource,
					TransferEvent.TRANSFER_PROGRESS, TransferEvent.REQUEST_PUT);
			// results[0]: groupName, results[1]: remoteFilename.
			results = getClient().upload_file(null, fileLength,
					new UploadFileSender(input, transferEvent), fileExtName,
					metaList);
		} catch (Exception e) {
			throw new IOException("Upload file \"" + uploadFileName + "\"fails");
		}
		return results;
	}

	protected StorageClient getClient() {
		StorageServer storageServer = null;
		return new StorageClient1(trackerServer, storageServer);
	}

	private class DownloadFileReceiver implements DownloadCallback {
		private OutputStream out;
		private Resource resource;
		private TransferEvent transferEvent;

		public DownloadFileReceiver(Resource resource, OutputStream output,
				TransferEvent transferEvent) {
			this.out = output;
			this.resource = resource;
			this.transferEvent = transferEvent;
		}

		@Override
		public int recv(long file_size, byte[] data, int bytes) {
			try {
				out.write(data, 0, bytes);
				fireTransferProgress(transferEvent, data, bytes);
			} catch (IOException e) {
				fireTransferError(resource, e, TransferEvent.REQUEST_GET);
				String msg = "GET request of: " + resource.getName() + " from "
						+ repository + " failed";

				throw new TransferFailedException(msg, e);
			}
			return 0;
		}

	}

	private class UploadFileSender implements UploadCallback {
		private InputStream input;
		private TransferEvent transferEvent;

		public UploadFileSender(InputStream input, TransferEvent transferEvent) {
			this.input = input;
			this.transferEvent = transferEvent;
		}

		public int send(OutputStream out) throws IOException {
			byte[] buffer = new byte[DEFAULT_BUFFER_SIZE];
			while (true) {
				int n = input.read(buffer);
				if (n == -1) {
					break;
				}
				out.write(buffer);
				fireTransferProgress(transferEvent, buffer, n);
			}
			return 0;
		}
	}

	public static class InvalidFastDfsResourceName extends Exception {
		private static final long serialVersionUID = -1709686325714569676L;

		public InvalidFastDfsResourceName() {
			super();
		}

		public InvalidFastDfsResourceName(String message) {
			super(message);
		}

		public InvalidFastDfsResourceName(String message, Throwable cause) {
			super(message, cause);
		}

		public InvalidFastDfsResourceName(Throwable cause) {
			super(cause);
		}
	}

	public static class FastDfsResourceName implements Serializable {
		private static final long serialVersionUID = 7275157175592890375L;

		private String groupName;
		private String remoteFilename;

		public FastDfsResourceName() {

		}

		public FastDfsResourceName(String resourceName)
				throws InvalidFastDfsResourceName {
			String[] names = resourceName.split(NAME_SEPARATOR);
			if (names.length != 2) {
				throw new InvalidFastDfsResourceName(
						"Invalid dfs resouce name.");
			}
			groupName = names[0];
			remoteFilename = names[1];
		}

		public FastDfsResourceName(String[] names) {
			groupName = names[0];
			remoteFilename = names[1];
		}

		public static FastDfsResourceName parse(String resourceName)
				throws InvalidFastDfsResourceName {
			return new FastDfsResourceName(resourceName);
		}

		public String getGroupName() {
			return groupName;
		}

		public void setGroupName(String groupName) {
			this.groupName = groupName;
		}

		public String getRemoteFilename() {
			return remoteFilename;
		}

		public void setRemoteFilename(String remoteFilename) {
			this.remoteFilename = remoteFilename;
		}

	}
}
